package cn.tianyigps.obd;

import cn.tianyigps.obd.beans.StandardModel;
import cn.tianyigps.obd.template.KafkaTemplate;
import cn.tianyigps.obd.template.MongoTemplate;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.client.MongoCollection;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * Created with IDEA
 *
 * @description: 存储操作类
 * @author: cui.guibin
 * @create: 2019/5/10 11:46
 **/
public class Storer {

    static Logger log = LogManager.getLogger(Storer.class);

    public static void store(String type,List<StandardModel> standardModels) {
        switch (type){
            case "1":
                store2Mongo(standardModels);
                break;
            case "2":
                store2Kafka(standardModels);
                break;
            case "3":
                store2Mongo(standardModels);
                store2Kafka(standardModels);
                break;
                default:
        }
    }

    public static void store2Disk(String hexStr, String flag, String path) {
        try {
            FileWriter writer=new FileWriter(path,true);
            SimpleDateFormat format=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time=format.format(new Date());
            writer.write(time + " " +flag + " " + hexStr + System.getProperty("line.separator"));
            writer.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void store2Mongo(List<StandardModel> standardModels) {
        MongoCollection<StandardModel> collection = MongoTemplate.getCollection("standardModel", StandardModel.class);
        collection.insertMany(standardModels);
    }
    public static void store2Kafka(List<StandardModel> standardModels) {
        // 转换成json字符串，再转换成DBObject对象
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        String json = null;
        try {
            json = objectMapper.writeValueAsString(standardModels);
        } catch (JsonProcessingException e) {
            log.error("标准对象序列化失败 ");
            log.error(e.getMessage());
            e.printStackTrace();
        }
        KafkaProducer<String, String> producer = KafkaTemplate.producer();
        producer.send(new ProducerRecord<>("inbound", json));
        producer.close();
    }


}
